package com.ruoyi.xdata.task;

import com.ruoyi.common.config.RuoYiConfig;
import com.ruoyi.common.constant.Constants;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.xdata.domain.XdataKettle;
import com.ruoyi.xdata.domain.XdataLog;
import com.ruoyi.xdata.service.*;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.core.logging.LoggingBuffer;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobEntryResult;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * 数据交换的定时任务
 */
@Slf4j
@Component("kettleTask")
public class KettleTask {

    @Autowired
    private IXdataKettleService xdataKettleService;
    @Autowired
    private IXdataLogService xdataLogService;

    /**
     * 执行一个数据通道
     * @param id
     */
    public int run(Long id){
        XdataKettle xdataKettle = xdataKettleService.selectXdataKettleById(id);
        if(xdataKettle == null){
            log.error("任务不存在： {}", id);
        }
        XdataLog xdataLog = new XdataLog();
        xdataLog.setKettleId(xdataKettle.getId());
        xdataLog.setKettleName(xdataKettle.getName());
        xdataLog.setStartTime(new Date());
        xdataLog.setResult("0");
        try{
            KettleEnvironment.init();
            EnvUtil.environmentInit();

            String filePath = RuoYiConfig.getProfile()+xdataKettle.getJobFile();
            log.info("开始执行kettle任务：{}", filePath);
            if(xdataKettle.getType().equalsIgnoreCase(XdataKettle.TYPE_KJB)){
                JobMeta jobMeta = new JobMeta(filePath, null);
                Job job = new Job(null, jobMeta);
                job.start();
                job.waitUntilFinished();
                if(job.getErrors() > 0){
                    xdataLog.setResult("1");
                    log(xdataLog, job.getJobEntryResults());
                }
            }else if(xdataKettle.getType().equalsIgnoreCase(XdataKettle.TYPE_KTR)){
                TransMeta transMeta = new TransMeta(filePath);
                Trans trans = new Trans(transMeta);
                trans.execute(null);
                trans.waitUntilFinished();
                if(trans.getErrors() > 0){
                    xdataLog.setResult("1");
                    trans.getActiveSubjobs().values().forEach(job -> {
                        log(xdataLog, job.getJobEntryResults());
                    });
                }
            }else{
                throw new ServiceException("不支持的任务类别");
            }
        }catch (Exception e){
            e.printStackTrace();
            xdataLog.setResult("1");
            xdataLog.setError(e.getMessage());
        }finally {
            xdataKettle.setLastTime(new Date());
            xdataKettleService.updateXdataKettle(xdataKettle);
        }
        xdataLog.setEndTime(new Date());
        xdataLogService.insertXdataLog(xdataLog);
        return 1;
    }

    /**
     * 解析日志
     * @param xdataLog
     * @param jobEntryResults
     */
    private void log(XdataLog xdataLog, List<JobEntryResult> jobEntryResults){
        LoggingBuffer loggingBuffer = KettleLogStore.getAppender();
        StringBuffer stringBuffer = new StringBuffer();
        jobEntryResults.forEach(jobEntryResult -> { //获取job任务中 多个转化出错的日志
            if(jobEntryResult.getResult().getNrErrors()>0){
                StringBuffer logTextBuffer = loggingBuffer.getBuffer(jobEntryResult.getLogChannelId(), false);
                stringBuffer.append(logTextBuffer);
            }
        });
        if(xdataLog.getError() == null){
            xdataLog.setError(stringBuffer.toString());
        }else{
            xdataLog.setError(xdataLog.getError()+"\r\n"+stringBuffer.toString());
        }
    }
}
